package com.mapreduce.reducejoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class TableMapper extends Mapper<LongWritable, Text,Text,TableBean> {
    private  String fileName;
    private Text k = new Text();
    private TableBean bean = new TableBean();
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // 获取输入文件切片
        FileSplit fileSplit = (FileSplit) context.getInputSplit();
       // 获取输入文件名称
        fileName = fileSplit.getPath().getName();

    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //读取每一行数据
        String line = value.toString();
        //处理不同类型的文件
        if (fileName.contains("order")){
            //截取
            String[] data = line.split("\t");
            k.set(data[1]); //以pid作为key
            bean.setOrderId(data[0]);
            bean.setpId(data[1]);
            bean.setAmount(Integer.parseInt(data[2]));
            bean.setpName("");
            bean.setFlag("order");
        }else {//产品表
            //截取
            String[] data = line.split("\t");
            k.set(data[0]); //以pid作为key
            bean.setpId(data[0]);
            bean.setpName(data[1]);
            bean.setOrderId("");
            bean.setAmount(0);
            bean.setFlag("pd");
        }

       context.write(k,bean);

    }
}
